package logMonitor;


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;
import logMonitor.bolt.MessageFilterBolt;
import logMonitor.bolt.ProcessMessage;
import logMonitor.bolt.TriggerMessage;
import logMonitor.spout.RandomSpout;
import logMonitor.spout.StringScheme;
import storm.kafka.BrokerHosts;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.ZkHosts;


/**
 * Created by lenovo on 2017/12/28.
 */
public class LogMonitorTopologyMain {
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InvalidTopologyException {
        TopologyBuilder builder = new TopologyBuilder();
        //集群===============================================
        //设置kafka的zookeeper集群l
        BrokerHosts hosts = new ZkHosts("192.168.223.201:2181,192.168.223.202:2181,192.168.223.203:2181");
        //初始化配置信息
        SpoutConfig spoutConfig = new SpoutConfig(hosts, "myOrder1", "/aaa", "log_monitor");
//        SpoutConfig spoutConfig = new SpoutConfig(brokerHosts, topic, zkRoot, spoutId);
        //在topology中设置spout
        builder.setSpout("kafka-spout", new KafkaSpout(spoutConfig),4);
        //本地===============================================
//        builder.setSpout("kafka-spout", new RandomSpout(new StringScheme()), 4);
        //<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<
        builder.setBolt("MessageFilter-bolt",new MessageFilterBolt(),4).shuffleGrouping("kafka-spout");
        //得到ip总流量排名
        builder.setBolt("ProcessMessage-bolt",new ProcessMessage(),4).shuffleGrouping("MessageFilter-bolt");
        //从规则表中取到想要监控的ip，将与该ip相关的记录存到redia,list,大key是ip，value是与大key有关的message
        builder.setBolt("TriggerMessage-bolt",new TriggerMessage(),4).shuffleGrouping("MessageFilter-bolt");
        Config topologConf = new Config();
        if (args != null && args.length > 0) {
            topologConf.setNumWorkers(4);
            StormSubmitter.submitTopologyWithProgressBar(args[0], topologConf, builder.createTopology());
        } else {
            topologConf.setMaxTaskParallelism(10);
            topologConf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 10);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("LogAnalyzeTopologyMain", topologConf, builder.createTopology());
            Utils.sleep(10000000);
            cluster.shutdown();
        }
    }
}
